package com.ysw.rpc.common.config;

import com.ysw.rpc.common.constants.RpcConstants;
import com.ysw.rpc.common.enums.MessageTypeEnum;
import com.ysw.rpc.common.messages.Message;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageCodec;

import java.util.List;

/**
 * @author YinShangwen
 * @since 2023/4/4 13:19
 */
@ChannelHandler.Sharable
public class MessageCodec extends MessageToMessageCodec<ByteBuf, Message> {
    private final RpcConfig rpcConfig;

    public MessageCodec(RpcConfig rpcConfig) {
        this.rpcConfig = rpcConfig;
    }

    @Override
    protected void encode(ChannelHandlerContext ctx, Message msg, List<Object> out) {
        Serializer.TypeMap typeMap = Serializer.TypeMap.get(rpcConfig.getSerializer());
        byte[] bytes = typeMap.getType().ser(msg);
        ByteBuf buffer = ctx.alloc().buffer();
        buffer
                .writeShort(RpcConstants.MAGIC_NUMBER)
                .writeInt(rpcConfig.getVersion())
                .writeByte(typeMap.ordinal())
                .writeByte(MessageTypeEnum.get(msg.getClass()).ordinal())
                .writeInt(bytes.length)
                .writeBytes(bytes);
        out.add(buffer);
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) {
        short magicNumber = msg.readShort();
        if (magicNumber != RpcConstants.MAGIC_NUMBER) {
            // discard invalid message
            return;
        }
        int version = msg.readInt();
        byte serializerOrdinal = msg.readByte();
        byte messageTypeOrdinal = msg.readByte();
        int dataLength = msg.readInt();
        byte[] bytes = new byte[dataLength];
        msg.readBytes(bytes);
        Message message = Serializer.TypeMap.values()[serializerOrdinal].getType()
                .deser(bytes, MessageTypeEnum.values()[messageTypeOrdinal].getClazz());
        out.add(message);
    }
}
